package io.kimmking.chatoom.client.consumer;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import io.kimmking.chatoom.client.config.Constant;
import io.kimmking.common.rocketmq.base.BaseIConsumer;
import io.kimmking.common.rocketmq.base.RocketMQMsgListener;
import io.kimmking.dubbo.common.entity.CharRoomMessage;
import io.kimmking.dubbo.common.service.CharRoomMessageService;
import io.kimmking.dubbo.common.unit.ThrowableActionV2;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.springframework.stereotype.Service;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

/**
 * 去消费MQ队列，将消费到的消息持久化存储(存储到Mysql)
 */
@Slf4j
@Service
@RocketMQMsgListener(
        //下边可以去掉，都使用的是默认值
        consumeMode = ConsumeMode.CONCURRENTLY, //默认值，并发接收模式
        messageModel = MessageModel.CLUSTERING, //默认值，集群模式
        selectorType = SelectorType.TAG, // 默认值，标签
        selectorExpression = "*" //默认值，匹配该topic下所有tag
)
public class ConsumerMsg extends BaseIConsumer<CharRoomMessage> {

    private final String topic = Constant.send_msg_topic;

    private final String cosumer_group = Constant.consumer_msg_group;

    @DubboReference(version = "1.0.0")
    CharRoomMessageService charRoomMessageService;

    @Override
    protected void handle(List<MessageExt> msgs) {
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        ArrayList<CharRoomMessage> charRoomMessages = Lists.newArrayList();
        msgs.stream()
                .filter(messageExt -> isVerify(messageExt, CharRoomMessage.class))
                .forEach(messageExt -> ThrowableActionV2.execute(()->{
                    CharRoomMessage charRoomMessage = JSONObject.parseObject(new String(messageExt.getBody(),
                            Charset.defaultCharset()), CharRoomMessage.class);
                    charRoomMessages.add(charRoomMessage);
                }));
        charRoomMessageService.saveBatch(charRoomMessages, 10000);
        return null;
    }
}
